#ifndef _TASK_H_
#define _TASK_H_
#include "flv_main.h"
#include "data_struct.h"
#include <pthread.h>
#define FLV_PORT        8080
#define PUSH_FLV_PORT   6666
#define PUSH_TASK_NUM 2
#define MAX_FLV_EVENT_NUMBER 6553
#define MAX_PUSH_EVENT_NUMBER 1000
#define MAX_FLV_EPOLL_NUM    6553
class task
{
public :
    pthread_t    listen_task;
    pthread_t    recv_stream_task;
    pthread_t    push_stream_task[PUSH_TASK_NUM];
    static data_struct *queue_handle;
    static struct epoll_event events[MAX_FLV_EVENT_NUMBER];
    static struct epoll_event push_events[MAX_PUSH_EVENT_NUMBER];
    static oal_uint32 epollfd;//在静态成员函数中可以直接访问静态成员变量
    static oal_uint32 recv_epollfd;
    static oal_uint32 listen_fd;//监听拉流
    static oal_uint32 push_listen_fd;//监听推流
    static oal_int32  pipe_fd[2];
    task();
    static oal_uint32 setnonblocking(oal_uint32 fd)
    {
	    oal_uint32 old_opt = fcntl(fd, F_GETFL);
	    oal_uint32 new_opt = old_opt | O_NONBLOCK;
	    fcntl(fd, F_SETFL, new_opt);
	    return old_opt;
    }
    static void addfd(oal_uint32 epoll_fd,oal_uint32 fd, oal_uint8 oneshot_flag)
    {
	    struct epoll_event event;
	    event.data.fd = fd;
	    event.events = EPOLLIN | EPOLLRDHUP | EPOLLHUP | EPOLLERR |EPOLLET;//边沿触发
	    if (oneshot_flag)
	    {
		    event.events = event.events | EPOLLONESHOT;//EPOLLONESHOT表示事件只能被一个线程处理
	    }
	    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event);
	    setnonblocking(fd);
    }
    static void modfd(oal_uint32 epoll_fd , oal_uint32 fd, oal_uint8 oneshot_flag, oal_uint32 ev)
    {
	    struct epoll_event event;
	    event.data.fd = fd;
	    event.events = ev | EPOLLRDHUP | EPOLLHUP | EPOLLERR | EPOLLET;
	    if (oneshot_flag)
	    {
		    event.events = event.events | EPOLLONESHOT;
	    }
	    epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &event);
    }
    static void delfd(oal_uint32 epoll_fd , oal_uint32 fd)
    {
        struct epoll_event event;
	    event.data.fd = fd;
	    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
    }
    static void  close_fd(oal_uint32 epoll_fd ,oal_uint32 fd)
    {
	    printf("fd close ,[%u]\n",fd);
	    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, 0);
	    close(fd);
    } 
    // static void create_push_listen_fd()
    // {
    //     push_listen_fd = socket(PF_INET, SOCK_STREAM, 0);
    //     if(push_listen_fd < 0)
    //     {
    //         printf("create listen socket failed\n");
    //         return ;
    //     }
    //     //优雅关闭连接
	//     struct linger tmp = { 0, 1 };
	//     setsockopt(push_listen_fd, SOL_SOCKET, SO_LINGER, &tmp, sizeof(tmp));
    //     struct sockaddr_in address;
    //     bzero(&address);
    //     address.sin_family = AF_INET;
	//     address.sin_addr.s_addr = htonl(INADDR_ANY);
	//     address.sin_port = htons(PUSH_FLV_PORT);
    //     int flag = 1;
	//     setsockopt(push_listen_fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));//不同ip端口可重复使用
    //     socklen_t len = sizeof(address);
    //     int ret = bind(push_listen_fd , (struct sockaddr *)&address,len);
    //     if(ret < 0)
    //     {
    //         printf("bind the addr/port failed\n");
    //         return ;
    //     }
    //     ret = listen(push_listen_fd, 5);//第二个参数为 TCP 未处理连接的队列长度，listen 非阻塞，accept
	//     if (ret < 0)
	//     {
	// 	    printf(" listen socket listen failed\n");
    //         return ;
	//     }
    //     addfd(push_listen_fd, 0);
    //     return ;
    // }
//     如果对描述符socket注册了EPOLLONESHOT事件，那么操作系统最多触发其上注册的一个可读、可写或者异常事件，且只触发一次。。想要下次再触发则必须使用epoll_ctl重置该描述符上注册的事件，包括EPOLLONESHOT 事件。

// EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里 
    static void create_listen_fd(oal_uint32 &listen_fd , oal_uint16 port)
    {
        listen_fd = socket(PF_INET, SOCK_STREAM, 0);
        printf("listen_fd is [%d]\n",listen_fd);
        if(listen_fd < 0)
        {   
            printf("create listen socket failed\n");
            return ;
        }
        //优雅关闭连接
	    struct linger tmp = { 0, 1 };
	    setsockopt(listen_fd, SOL_SOCKET, SO_LINGER, &tmp, sizeof(tmp));
        struct sockaddr_in address;
        bzero(&address,sizeof(struct sockaddr_in));
        address.sin_family = AF_INET;
	    address.sin_addr.s_addr = htonl(INADDR_ANY);
	    address.sin_port = htons(port);
        int flag = 1;
	    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));//不同ip端口可重复使用
        socklen_t len = sizeof(address);
        int ret = bind(listen_fd , (struct sockaddr *)&address,len);
        if(ret < 0)
        {
            printf("bind the addr/port failed\n");
            return ;
        }
        ret = listen(listen_fd, 5);//第二个参数为 TCP 未处理连接的队列长度，listen 非阻塞，accept
	    if (ret < 0)
	    {
		    printf(" listen socket listen failed\n");
		    return ;
	    }
	    addfd(epollfd,listen_fd, 0);
        return ;
    }
    static void * listen_func(void* arg)
    {
        printf("listen_func on\n");
        epollfd = epoll_create(MAX_FLV_EPOLL_NUM);//参数MAX_EPOLL_NUM 是epoll监听的最大socket id 数
        printf("epollfd is [%d]\n",epollfd);
	    if (epollfd < 0)
	    {
		    printf(" epoll_fd failed\n");
		    return NULL;
	    }
        create_listen_fd(listen_fd , FLV_PORT);
        create_listen_fd(push_listen_fd , PUSH_FLV_PORT);
        //oal_int32 recv_epollfd1 = epoll_create(666);//参数MAX_EPOLL_NUM 是epoll监听的最大socket id 数
        //oal_int32 recv_epollfd2 = epoll_create(666);//参数MAX_EPOLL_NUM 是epoll监听的最大socket id 数
        //printf("recv_epollfd1 is [%d] ,recv_epollfd2 [%d]\n",recv_epollfd1,recv_epollfd2);
        while (1)
	    {
		    //printf("epoll wait!!!\n");
		    int num = epoll_wait(epollfd, events, MAX_FLV_EVENT_NUMBER, -1);//最后一个参数表示超时返回时间，超时返回0
            printf("listen_func epoll wait on!!!\n");
		    if(num < 0 && errno == EINTR )
		    {
			    printf("interrupt!!!\n");
			    continue;
		    }
		    for (int i = 0; i < num; i++)
		    {
			    if (events[i].data.fd == listen_fd && (EPOLLIN&events[i].events) )
			    {
				    struct sockaddr_in address;
				    socklen_t cli_len = sizeof(address);
				    oal_int32 tempfd = accept(listen_fd,(struct sockaddr *) &address,&cli_len);
				    if (tempfd < 0)
				    {
					    printf("accept failed\n");
					    continue;
				    }
                    oal_int32 userfd = queue_handle->get_user_id();
                    if(userfd < 0)
                    {
                        printf("get userid wrong!!\n");
                        continue;
                    }
                    printf("new user in\n");
                    queue_handle->user_head[userfd].user_clear();
                    queue_handle->user_head[userfd].private_id = userfd;
                    queue_handle->user_head[userfd].socket_id  = tempfd;
                    queue_handle->fd2user_map->insert(pair<oal_uint32,oal_uint32>(tempfd,userfd));
                    addfd(epollfd,tempfd, 1);
			    }
                else if(events[i].data.fd == push_listen_fd && (EPOLLIN&events[i].events) )
                {
                    printf("push in , listen fd [%d]\n",events[i].data.fd);
                    oal_int8 p = 1;
                    write(pipe_fd[1], &p, sizeof(p));
                }
                /*  这一步没有必要，因为探测连接关闭可以在推送的时候
			    else if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))//出现 连接关闭,错误，描述符被挂起
			    {
				    printf("epoll close\n");
				    oal_int32 temp_fd = events[i].data.fd;
                    oal_int32 real_fd = fd2entry_map[temp_fd];

			    }
                */
			    else if (events[i].events & EPOLLIN)
			    {
                    //printf("epoll in\n");
                    oal_int32 read_fd  = events[i].data.fd;
                    printf("epoll in ,fd [%d]\n",read_fd);
                    oal_uint32 user_id = queue_handle->fd2user_map->find(read_fd)->second;
                    if(queue_handle->user_head[user_id].read() > 0)//执行http状态机,读完之后就将描述符从epoll那里删除，因为描述符只需要可写
                    {
                        if(queue_handle->pusher2enrtry_map->find(queue_handle->user_head[user_id].puser_id) != queue_handle->pusher2enrtry_map->end())
                        {
                            oal_int32 entry_id = queue_handle->pusher2enrtry_map->find(queue_handle->user_head[user_id].puser_id)->second;
                            pthread_spin_lock(&(queue_handle->entry_head[entry_id].userlist_spinlock)); //应该用互斥锁
                            queue_handle->entry_head[entry_id].user_list->push_back((oal_int32)user_id);//将用户加入到推流实体中
                            pthread_spin_unlock(&(queue_handle->entry_head[entry_id].userlist_spinlock)); //应该用互斥锁
                            delfd(epollfd,read_fd);
                        }
                        else
                        {
                            queue_handle->user_head[user_id].write_error();
                            queue_handle->free_user_id(user_id);
                            close_fd(epollfd,read_fd);
                        }
                    }
                    else
                    {
                        modfd(epollfd ,read_fd, 1, EPOLLIN);
                    }

			    }
			    else
			    {
				    printf("something happened unexpected!!\n");
			    }
		    }
	    }
        return NULL;
    }
	static void * recv_stream_func(void* arg)
    {
        printf("recv_stream_func on\n");
        recv_epollfd = epoll_create(MAX_FLV_EPOLL_NUM);//参数MAX_EPOLL_NUM 是epoll监听的最大socket id 数
        //oal_int32 recv_epollfd1 = epoll_create(666);//参数MAX_EPOLL_NUM 是epoll监听的最大socket id 数
        printf("recv_epollfd is [%d] \n",recv_epollfd);
	    if (recv_epollfd < 0)
	    {
		    printf(" epoll_fd failed\n");
		    return NULL;
	    }
        addfd(recv_epollfd,pipe_fd[0],0);
        printf("recv ,the pipe  fd is [%d] [%d]\n",pipe_fd[0],pipe_fd[1]);
        while(1)
        {
            oal_int32 num = epoll_wait(recv_epollfd, push_events, MAX_PUSH_EVENT_NUMBER,-1);//每0.5s取一次环
            oal_int32 ret = BUFFER_GET_SUCCESS;
            oal_int32 socketid;
            //printf(" recv_stream_func epoll_fd on, event num is [%d]\n",num);
            while(0)
            {
                socketid = 0;
                ret = buffer_get(&queue_handle->push_socket_buffer,(oal_uint8 *)&socketid);
                if(ret != BUFFER_GET_SUCCESS)
                {
                    break;
                }
                printf("get one push \n");
                oal_int32 pusherfd = queue_handle->get_stream_id();
                if(pusherfd < 0)
                {
                    printf("get  pusherfd wrong!!\n");
                    continue;
                }
                queue_handle->entry_head[pusherfd].socket_id   = socketid;
                queue_handle->entry_head[pusherfd].private_id  = pusherfd;
                //queue_handle->fd2entry_map->insert(pair<oal_uint32,oal_uint32>(socketid,pusherfd));//加速
                queue_handle->insert_entrymap(socketid,pusherfd);
                addfd(recv_epollfd,socketid, 1);
            }
            
            for(oal_int32 i = 0 ; i < num ; i++)
            {
                if (push_events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))//出现 连接关闭,错误，描述符被挂起
			    {
				    printf("recv epoll close\n");
                    oal_int32 temp_fd = push_events[i].data.fd;
                    oal_int32 real_fd = queue_handle->fd2entry_map->find(temp_fd)->second;
				    queue_handle->entry_head[real_fd].is_close = true;
                    queue_handle->put_sched_entry(real_fd);//连接关闭时通知flv实体去取完数据
                    queue_handle->delete_entrymap(temp_fd);
                    close_fd(recv_epollfd,temp_fd);
			    }
                else if(push_events[i].data.fd == pipe_fd[0] && (events[i].events & EPOLLIN))
                {
                    struct sockaddr_in address;
				    socklen_t cli_len = sizeof(address);
				    oal_int32 tempfd = accept(push_listen_fd,(struct sockaddr *) &address,&cli_len);
				    if (tempfd < 0)
				    {
					    printf("accept failed\n");
					    continue;
				    }
                    printf("accept tempfd is [%d]\n",tempfd);
                    oal_int32 pusherfd = queue_handle->get_stream_id();
                    queue_handle->entry_head[pusherfd].init();
                    queue_handle->entry_head[pusherfd].socket_id   = tempfd;
                    queue_handle->entry_head[pusherfd].private_id  = pusherfd;
                    //queue_handle->fd2entry_map->insert(pair<oal_uint32,oal_uint32>(socketid,pusherfd));//加速
                    queue_handle->insert_entrymap(tempfd,pusherfd);
                    addfd(recv_epollfd,tempfd, 1); 
                }
			    else if (push_events[i].events & EPOLLIN)
			    {
                    //printf("push epoll in\n");
                    oal_int32 temp_fd = push_events[i].data.fd;
                    //printf("1111,the push socket fd [%d]\n",temp_fd);
                    oal_int32 real_fd = queue_handle->fd2entry_map->find(temp_fd)->second;
                    //printf("2222\n");
                    //printf("1111,the push real_fd [%d]\n",real_fd);
                    queue_handle->entry_head[real_fd].read();
                    printf("3333\n");
                    //sleep(100);
                    if(!queue_handle->entry_head[real_fd].find_mysql_flag)
                    {
                        queue_handle->pusher2enrtry_map->insert(pair<oal_uint32,oal_uint32>(queue_handle->entry_head[real_fd].push_user_id,real_fd));
                        queue_handle->entry_head[real_fd].find_mysql_flag = true;
                    }
                    queue_handle->put_sched_entry(real_fd);
                     printf("444\n");
                    modfd(recv_epollfd ,temp_fd, 1, EPOLLIN);
                    
                }
            }
           
        }
        return NULL;
    }
    static void * push_stream_func(void* arg)
    {
        printf("push stream on\n");
        while(1)
        {
            oal_int32 entry_id = queue_handle->get_sched_entry();
            stream_entry &entry_obj = queue_handle->entry_head[entry_id];
            if(entry_obj.push_stream(queue_handle) < 0) //推流完成
            {
                entry_obj.stream_entry_free();
                queue_handle->free_stream_id(entry_obj.private_id);
            }
        }
         return NULL;
    }
};
#endif